{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "4310eebd",
   "metadata": {},
   "source": [
    "## 协程\n",
    "\n",
    "子程序，或者称为函数，在所有语言中都是层级调用，比如A调用B，B在执行过程中又调用了C，C执行完毕返回，B执行完毕返回，最后是A执行完毕。所以子程序调用是通过栈实现的，一个线程就是执行一个子程序\n",
    "\n",
    "子程序调用总是一个入口，一次返回，调用顺序是明确的。而协程的调用和子程序不同\n",
    "\n",
    "协程，又称微线程，纤程。英文名Coroutine\n",
    "\n",
    "协程看上去也是子程序，但执行过程中，在子程序内部可中断，然后转而执行别的子程序，在适当的时候再返回来接着执行\n",
    "\n",
    "> 注意，在一个子程序中中断，去执行其他子程序，不是函数调用，有点类似CPU的中断。比如子程序A、B：\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fe701ba8",
   "metadata": {},
   "source": [
    "## 可以实现协程的方式\n",
    "\n",
    "1. greenlet\n",
    "2. yield关键字\n",
    "3. asyncio\n",
    "4. async & await"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d87ebba2",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "\n",
    "async def fun1():\n",
    "    print(1)\n",
    "    await asyncio.sleep(2)\n",
    "    print(2)\n",
    "    return 'a'\n",
    "\n",
    "async def fun2():\n",
    "    print(3)\n",
    "    await asyncio.sleep(2)\n",
    "    print(4)\n",
    "    return 'b'\n",
    "\n",
    "tasks = [\n",
    "    asyncio.ensure_future( fun1() ),\n",
    "    asyncio.ensure_future( fun2() )\n",
    "]\n",
    "\n",
    "async def main():\n",
    "    p1,p2 = await asyncio.wait(tasks) # asyncio.wait并不会返回 协程函数\n",
    "    print(p1)\n",
    "    print(p2)\n",
    "\n",
    "await main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5aa60c00",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "\n",
    "asyncio.create_task\n",
    "\n",
    "async def fun1():\n",
    "    print(1)\n",
    "    await asyncio.sleep(2)\n",
    "    print(2)\n",
    "    return 'a'\n",
    "\n",
    "async def fun2():\n",
    "    print(3)\n",
    "    await asyncio.sleep(2)\n",
    "    print(4)\n",
    "    return 'b'\n",
    "\n",
    "async def main():\n",
    "    tasks = [\n",
    "        asyncio.create_task(fun1()),\n",
    "        asyncio.create_task(fun2())\n",
    "    ]\n",
    "    p1, p2 = await asyncio.wait(tasks)\n",
    "    print(type(p1)) # set ?\n",
    "    print(p2)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    asyncio.run(main())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c386bb51",
   "metadata": {},
   "source": [
    "## python几种可等待的对象\n",
    "\n",
    "- 协程对象 (即 async前缀的函数调用后返回的协程对象)\n",
    "- Task对象 （asyncio的Task对象）\n",
    "- Future对象 （asyncio的底层一点的Future对象）"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9f81a20b",
   "metadata": {},
   "source": [
    "## python中的另一个future\n",
    "\n",
    "来自于多线程多进程相关的concurrent库下的名字也叫future的类"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f5bf3cad",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import time\n",
    "from concurrent.futures import Future\n",
    "from concurrent.futures.thread import ThreadPoolExecutor\n",
    "from concurrent.futures.process import ProcessPoolExecutor\n",
    "\n",
    "def func(value):\n",
    "    time.sleep(1)\n",
    "    print(value)\n",
    "    return 123\n",
    "\n",
    "# pool = ThreadPoolExecutor(max_workers=5)\n",
    "pool = ProcessPoolExecutor(max_workers=5)\n",
    "\n",
    "for i in range(10):\n",
    "    fut = pool.submit(func, i) # 这里返回的 fut 也也是 Futrue，注意不是asyncio里面的Future\n",
    "    print(type(fut))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b9204df7",
   "metadata": {},
   "source": [
    "## run_in_executor解决当前进程使用asyncio但是三方库不支持asyncio的问题"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "006a8a0e",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "RuntimeError",
     "evalue": "asyncio.run() cannot be called from a running event loop",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mRuntimeError\u001b[0m                              Traceback (most recent call last)",
      "Input \u001b[1;32mIn [2]\u001b[0m, in \u001b[0;36m<cell line: 37>\u001b[1;34m()\u001b[0m\n\u001b[0;32m     35\u001b[0m     \u001b[38;5;28mprint\u001b[39m(r2)\n\u001b[0;32m     37\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;18m__name__\u001b[39m \u001b[38;5;241m==\u001b[39m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124m__main__\u001b[39m\u001b[38;5;124m'\u001b[39m:\n\u001b[1;32m---> 38\u001b[0m     asyncio\u001b[38;5;241m.\u001b[39mrun(main())\n",
      "File \u001b[1;32mD:\\apps\\Python39\\lib\\asyncio\\runners.py:33\u001b[0m, in \u001b[0;36mrun\u001b[1;34m(main, debug)\u001b[0m\n\u001b[0;32m      9\u001b[0m \u001b[38;5;124;03m\"\"\"Execute the coroutine and return the result.\u001b[39;00m\n\u001b[0;32m     10\u001b[0m \n\u001b[0;32m     11\u001b[0m \u001b[38;5;124;03mThis function runs the passed coroutine, taking care of\u001b[39;00m\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m     30\u001b[0m \u001b[38;5;124;03m    asyncio.run(main())\u001b[39;00m\n\u001b[0;32m     31\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[0;32m     32\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m events\u001b[38;5;241m.\u001b[39m_get_running_loop() \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[1;32m---> 33\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mRuntimeError\u001b[39;00m(\n\u001b[0;32m     34\u001b[0m         \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124masyncio.run() cannot be called from a running event loop\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m     36\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m coroutines\u001b[38;5;241m.\u001b[39miscoroutine(main):\n\u001b[0;32m     37\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124ma coroutine was expected, got \u001b[39m\u001b[38;5;132;01m{!r}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39mformat(main))\n",
      "\u001b[1;31mRuntimeError\u001b[0m: asyncio.run() cannot be called from a running event loop"
     ]
    }
   ],
   "source": [
    "import time\n",
    "import asyncio\n",
    "import concurrent.futures\n",
    "from random import randint\n",
    "\n",
    "def func1(value):\n",
    "    time.sleep(randint(1, 2))\n",
    "    print(value)\n",
    "    return value\n",
    "\n",
    "async def main():\n",
    "    loop = asyncio.get_running_loop()\n",
    "\n",
    "    futs = []\n",
    "    # for i in range(10):\n",
    "    #     # 1. 内部会先调用 ThreadPoolExecutor的submit方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象\n",
    "    #     # 2. 调用asyncio.wrap_future对象将concurrent.futures.Future包装为asyncio.Future对象。这样就能支持await关键字\n",
    "    #     fut = loop.run_in_executor(None, func1, i)\n",
    "    #     futs.append(fut)\n",
    "    # r1, r2 = await asyncio.wait(futs)\n",
    "    # print(r1, r2)\n",
    "\n",
    "    # with concurrent.futures.ThreadPoolExecutor() as pool:\n",
    "    #     for i in range(10):\n",
    "    #         futs.append(loop.run_in_executor(pool, func1, i))\n",
    "\n",
    "    with concurrent.futures.ProcessPoolExecutor() as pool:\n",
    "        for i in range(10):\n",
    "            futs.append(loop.run_in_executor(pool, func1, i))\n",
    "    r1, r2 = await asyncio.wait(futs)\n",
    "    print(r1)\n",
    "    print('---------------')\n",
    "    for f in r1:\n",
    "        print(111, f.result())\n",
    "    print(r2)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    asyncio.run(main())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2b09c5bb",
   "metadata": {},
   "source": [
    "## uvloop\n",
    "``` python\n",
    "import asyncio\n",
    "import uvloop\n",
    "\n",
    "asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())\n",
    "\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b23af192",
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import uvloop\n",
    "\n",
    "asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fa409c69",
   "metadata": {},
   "source": [
    "## 示例1 利用已经支持异步的asioredis库操作redis\n",
    "\n",
    "使用 aioredis 操作redis\n",
    "\n",
    "``` pip install aioredis ```\n",
    "\n",
    "``` python\n",
    "import asyncio\n",
    "import aioredis\n",
    "\n",
    "async def execute(address, password):\n",
    "    \n",
    "    redis = await aioredis.create_redis_pool(address, password=password)\n",
    "\n",
    "    await redis.hmset_dict('car', key1=1, key2=2, key3=3)\n",
    "\n",
    "    result = await redis.hgetall('car', encoding='utf-8')\n",
    "\n",
    "    redis.close()\n",
    "\n",
    "    await redis.wait_close()\n",
    "\n",
    "task_list = [\n",
    "    execute('redis://11.111.111', 'rott:abc'),\n",
    "    execute('redis://11.111.111', 'rott:abc')\n",
    "]\n",
    "\n",
    "asyncio.run(asyncio.wait(task_list))\n",
    "# 问？：为什么不在task_list中直接创建Task，而是在asyncio.wait中创建\n",
    "# 因为create_task需要在asyncio.run之后运行，否则不会被加入到事件循环中\n",
    "\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22f0bf69",
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "\n",
    "async def async_hello(future):\n",
    "    print('async hello')\n",
    "    await asyncio.sleep(2)\n",
    "    future.set_result('Hello World') # future对象的set_result方法\n",
    "\n",
    "def hello():\n",
    "    ft = asyncio.Future()\n",
    "    asyncio.ensure_future(async_hello(ft)) # ensure_future里的参数如果是Task warpper或者协程对象，这个对象讲安排调度\n",
    "    # 重点就是这里，返回一个Future\n",
    "    return ft\n",
    "\n",
    "async def run():\n",
    "    print('Run ...')\n",
    "    r = await hello()\n",
    "    print(r)\n",
    "\n",
    "loop = asyncio.get_event_loop()\n",
    "loop.run_until_complete(run())\n",
    "loop.close()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a2af9c65",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
